package com.iceland.common.listener;

import java.util.concurrent.*;

public class ConsumeQueue<T> extends LinkedBlockingQueue<T>{

    private LinkedBlockingQueue<ConsumeListener> listenerQueue=new LinkedBlockingQueue<>();

//    private static ThreadPoolExecutor threadPoolExecutor= new ThreadPoolExecutor(2,2,10, TimeUnit.SECONDS,new ArrayBlockingQueue<>(500),new MessageThreadFactory("MessageEventExcutor"));
    public int getListenerSize(){
        return listenerQueue.size();
    }
    Thread thread;
    public ConsumeQueue(){
        thread=new Thread(new Runnable() {
            @Override
            public void run() {
                while (true){
                    try {
                        //阻塞获取listener
                        ConsumeListener listener = listenerQueue.take();
                        T element  =null;
                        int timeout=1;
                        while (!listener.isExpire()){
                            //阻塞获取element
                            element=ConsumeQueue.this.poll(timeout,TimeUnit.MILLISECONDS);
                            if(timeout<1000){
                                timeout*=2;
                            }
                            if(element!=null){
                                timeout=1;
                                break;
                            }
                        }
                        if(listener.isExpire() && !expire(element,listener)){
                            listener.expire(element);
                            continue;
                        }

                        if(!beforeConsume(element,listener)){
                            continue;
                        }
                        listener.consume(element);
                        afterConsume(element,listener);

                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        try {
                            Thread.sleep(3000);
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                    }


                }

            }

        });
        thread.setName("ListenerQueue-Thread");
        thread.start();
    }

    /**
     * 消费执行前
     * @param element
     * @param listener
     * @return 是否执行consume
     */
    public boolean beforeConsume(T element, ConsumeListener listener) {
        return true;
    }

    public boolean expire(T element, ConsumeListener listener) {
        return false;
    }

    /**
     * 消费执行后
     * @param element
     * @param listener
     * @return
     */
    public void afterConsume(T element, ConsumeListener listener) {

    }

    /**
     * 添加新的元素
     */
    public void produce(T element){
        this.add(element);
    }

    /**
     * 添加监听器
     * @param listener
     */
    public synchronized void addListener(ConsumeListener listener){
        listener.updateExpireTime();
        listenerQueue.add(listener);
    }



}
